package com.gitee.ywj1352.reactive;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.reactivestreams.Subscription;
import reactor.core.Exceptions;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.SignalType;

import java.io.ByteArrayInputStream;
import java.io.ObjectInputStream;

public class ZKBusinessSubscriber extends BaseSubscriber<DistributedEventObject> {

    private int max;

    Subscription subscription;

    public int getMax() {
        return max;
    }

    public ZKBusinessSubscriber(int max) {
        this.max = max;

        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.128.129:2181",
                5000, 5000, retryPolicy);
        client.start();

        CuratorCache curatorCache = CuratorCache.builder(client, "event").build();
        CuratorCacheListener listener = CuratorCacheListener
                .builder()
                .forPathChildrenCache("event", client, (c, cevent) -> {
                    if (PathChildrenCacheEvent.Type.CHILD_ADDED.equals(cevent.getType())){
                        // 添加 child 代表法布事件
                        byte[] data = cevent.getData().getData();
                        DistributedEventObject d = (DistributedEventObject)new ObjectInputStream(new ByteArrayInputStream(data)).readObject();
                        super.onNext(d);
                    }
                })
                .build();
        curatorCache.listenable().addListener(listener);
        curatorCache.start();


    }

    public void hookOnSubscribe(Subscription subscription) {
        System.out.println("BusinessSubscriber#hookOnSubscribe : "+subscription.getClass());
        this.subscription = subscription;
        subscription.request(max);
    }

    public void hookOnNext(DistributedEventObject value) {
        System.out.println("do on next " + value);

    }


    public void hookOnComplete() {
        System.out.println("do on complete");
    }


    public void hookOnError(Throwable throwable) {
        //
        throw Exceptions.errorCallbackNotImplemented(throwable);
    }


    public void hookOnCancel() {
        System.out.println("Cancel");
    }

    public void hookFinally(SignalType type) {
        System.out.println(type);
    }

}
